package com.hrvsr.observer.distributed;

import java.io.Closeable;
import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import org.apache.commons.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.hrvsr.observer.Observer;
import com.hrvsr.observer.State;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.Tuple;

public class RedisObserver implements Observer, Closeable {
	private static Logger logger = LoggerFactory.getLogger(RedisObserver.class);

	private JedisPool jedisPool;
	private int dbNo;
	private String keyPrefix;

	public RedisObserver(JedisPool jedisPool, int dbNo, String keyPrefix) {
		super();
		this.jedisPool = jedisPool;
		this.dbNo = dbNo;
		this.keyPrefix = keyPrefix;
	}

	@Override
	public boolean clustersDone() {
		Jedis jedis = null;
		try {
			jedis = jedisPool.getResource();
			jedis.select(dbNo);
			Set<Tuple> nodeStats = jedis.zrangeByScoreWithScores(keyPrefix + "Nodes", State.NOT_INITED, State.DONE);
			for (Tuple nodeStat : nodeStats) {
				double score = nodeStat.getScore();
				boolean idleOrDone = score == State.IDLE || score == State.DONE;
				if (!idleOrDone)
					return false;
			}

			logger.info("redis observer thinks culusters done");
//			if (nodeStats.size() <= 0) {
//				logger.info("but the nodeStats is empty, so return false");
//				return false;
//			}
//			
//			for (Tuple nodeStat : nodeStats) {
//				logger.info("node stat is: " + nodeStat.getScore());
//			}
			return true;
		} finally {
			IOUtils.closeQuietly(jedis);
		}
	}

	@Override
	public void update(State state) {
		String nodeId = state.getNodeId();
		Jedis jedis = null;
		try {
			jedis = jedisPool.getResource();
			jedis.select(dbNo);
			Transaction trans = jedis.multi();
			// if (state.getStat() == State.DONE) {
			// trans.zrem(keyPrefix + "Nodes", state.getNodeId());
			// trans.del(keyPrefix + state.getNodeId());
			// } else {
			trans.zadd(keyPrefix + "Nodes", state.getStat(), state.getNodeId());
			trans.hmset(keyPrefix + nodeId, stateToMap(state));
			// }
			trans.exec();
		} finally {
			IOUtils.closeQuietly(jedis);
		}
	}

	private Map<String, String> stateToMap(State state) {
		Map<String, String> map = new HashMap<String, String>();
		Field[] fields = state.getClass().getDeclaredFields();
		for (Field f : fields) {
			if (Modifier.isStatic(f.getModifiers()))
				continue;

			f.setAccessible(true);
			try {
				Object o = f.get(state);
				if (o != null) {
					map.put(f.getName(), o.toString());
				}
			} catch (IllegalArgumentException | IllegalAccessException e) {
				// ignore
			}
		}

		return map;
	}

	@Override
	public void close() throws IOException {
		IOUtils.closeQuietly(jedisPool);
	}

}
